package com.ai.nsa.inputer;

import com.ai.nsa.DefaultNSATableRoute;
import com.ai.nsa.NSACache;
import com.ai.nsa.NSASerial;
import com.ai.nsa.NSATableRoute;
import com.ai.nsa.utils.MapTools;
import com.ai.nsa.utils.hbase.HBaseOperations;
import com.ai.nsa.utils.hbase.model.HBaseRow;
import com.ai.nsa.utils.hbase.model.RowKey;
import com.ai.nsa.utils.partition.HashPartitioner;
import com.ai.nsa.utils.partition.Partitioner;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Map;

import static com.ai.nsa.NSAConstants.*;
import static com.ai.nsa.NSAFields.MASTER_FIELDS;

/**
 * @author Feng Chen 2017/6/19.
 */
public abstract class AbstractInputer implements Inputer {

    protected NSASerial serial;
    protected HBaseOperations hbase;
    protected NSACache cache;

    protected NSATableRoute tableRoute;

    protected static final SimpleDateFormat FULL_FORMAT = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    protected static final SimpleDateFormat DATE_FORMAT = new SimpleDateFormat("yyyyMMdd");

    private static final Logger logger = LoggerFactory.getLogger(AbstractInputer.class);

    public AbstractInputer(NSASerial serial, HBaseOperations hbase, NSACache cache) {
        this.serial = serial;
        this.hbase = hbase;
        this.cache = cache;

        this.tableRoute = new DefaultNSATableRoute();
    }

    @Override
    public void put(Map<String, Object> row) throws Exception {

        // 数据微调
        rename(row);

        // 生成 rowkey
        String rowkey = rowKey(row);

        if (rowkey == null || rowkey.length() == 0) {
            return;
        }

        // 插入主表
        putMaster(rowkey, row);

        // 插入业务表
        putSlave(rowkey, row);

        // 清理缓存
        clear(row);

    }

    /**
     * 数据微调
     *
     * @param row
     */
    public abstract void rename(Map<String, Object> row);

    /**
     * 生成rowKey
     *
     * @param row
     * @return
     * @throws Exception
     */
    public String rowKey(Map<String, Object> row) throws Exception {
        return new RowKeyCreater() {

            private Partitioner partitioner = new HashPartitioner();

            @Override
            public String get(Map<String, Object> event) throws Exception {
                RowKey rowKey = new RowKey();

                String serialNumber = MapTools.getString(event, COLUMN_SERIAL_NUMBER);
                if (StringUtils.isEmpty(serialNumber)) {
                    return null;
                }

                String partition = "" + partitioner.getPartition(serialNumber);

                String tradeDate = MapTools.getString(event, COLUMN_TRADE_DATE);
                if (StringUtils.isEmpty(tradeDate)) {
                    return null;
                }

                rowKey.part(partition)
                        .id(serialNumber)
                        .date(tradeDate)
                        .serial(serial.get(serialNumber, tradeDate));
                return rowKey.get();
            }

        }.get(row);
    }

    /**
     * 插入主表
     *
     * @param rowKey
     * @param row
     * @throws Exception
     */
    public void putMaster(String rowKey, Map<String, Object> row) throws Exception {
        HBaseRow hBaseRow = new Converter() {

            @Override
            public HBaseRow convert(Map<String, Object> event) {
                if (event == null || event.isEmpty()) {
                    return null;
                }

                HBaseRow row = new HBaseRow();

                // 设置 列组
                for (String key : event.keySet()) {

                    if (event.get(key) == null || !MASTER_FIELDS.contains(key)) {
                        continue;
                    }

                    row.addValue(HBASE_FAMILY, key, MapTools.getString(event, key));
                }

                return row;
            }

        }.convert(row);
        hBaseRow.setRowKey(rowKey);
        hbase.put(HBASE_MASTER_TABLE + UNDERLINE + tableRoute.getTable(rowKey), hBaseRow);
    }

    public String convert(String str) {

        if (StringUtils.isEmpty(str)) {
            return null;
        }

        try {
            Date date = FULL_FORMAT.parse(str);
            return DATE_FORMAT.format(date);
        } catch (ParseException e) {
            logger.error("", e);
            return null;
        }
    }

    /**
     * 插入业务表
     *
     * @param rowKey
     * @param row
     * @throws Exception
     */
    public abstract void putSlave(String rowKey, Map<String, Object> row) throws Exception;

    public void clear(Map<String, Object> row) throws Exception {
        String serialNumber = MapTools.getString(row, COLUMN_SERIAL_NUMBER);
        while (true) {
            String key = cache.pop(serialNumber);
            if (StringUtils.isEmpty(key)) {
                break;
            }
            cache.delete(key);
        }
    }


    /////////////////////////////////////////////////

    /**
     * hbase 行转换器
     */
    protected interface Converter {

        HBaseRow convert(Map<String, Object> event);

    }

    /**
     * rowkey 生成器
     */
    protected interface RowKeyCreater {
        String get(Map<String, Object> event) throws Exception;
    }

}
